package com.lollipop.rabbitmq.four;

import cn.hutool.core.lang.UUID;
import com.lollipop.rabbitmq.util.RabbitMqUtiils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

/**
 * @Author: Lollipop
 * @Date: 2021-10-05 15:22
 * @Description: 发布确认模式：1.单个确认 2.批量确认 3.异步批量确认
 */
public class ConfirmMessage {
    public static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        // 单个消息发布确认 耗时：573ms
        //ConfirmMessage.publishMessageIndividually();
        // 批量消息发布确认 耗时：42ms
        //ConfirmMessage.publishMessageBatch();
        // 异步消息发布确认 耗时：21ms
        ConfirmMessage.publishMessageAsync();
    }


    /**
     * 单个消息发布确认
     *
     * @throws IOException
     * @throws TimeoutException
     * @throws InterruptedException
     */
    public static void publishMessageIndividually() throws IOException, TimeoutException, InterruptedException {
        Channel channel = RabbitMqUtiils.getChannel();
        // 声明队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, false, false, null);
        // 开启发布确认模式
        channel.confirmSelect();

        Instant start = Instant.now();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));
            // 发布确认
            boolean flag = channel.waitForConfirms();
            if (flag) {
                System.out.println("消息发送成功！");
            }
        }
        Instant end = Instant.now();
        Duration duration = Duration.between(start, end);
        System.out.println("消息发布单个确认耗时：" + duration.toMillis() + "ms");
    }

    /**
     * 批量消息发布确认
     *
     * @throws IOException
     * @throws TimeoutException
     * @throws InterruptedException
     */
    public static void publishMessageBatch() throws IOException, TimeoutException, InterruptedException {
        Channel channel = RabbitMqUtiils.getChannel();
        // 声明队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, false, false, null);
        // 开启发布确认模式
        channel.confirmSelect();

        Instant start = Instant.now();

        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));

            // 批量确认消息大小
            int batchSize = 100;

            if (i % batchSize == 0) {
                // 发布确认
                boolean flag = channel.waitForConfirms();
                if (flag) {
                    System.out.println("消息发送成功！");
                }
            }
        }

        Instant end = Instant.now();
        Duration duration = Duration.between(start, end);
        System.out.println("消息发布批量确认耗时：" + duration.toMillis() + "ms");
    }

    /**
     * 异步发布确认
     *
     * @throws IOException
     * @throws TimeoutException
     * @throws InterruptedException
     */
    public static void publishMessageAsync() throws IOException, TimeoutException, InterruptedException {
        Channel channel = RabbitMqUtiils.getChannel();
        // 声明队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, false, false, null);
        // 开启发布确认模式
        channel.confirmSelect();

        // 线程安全有序哈希表
        ConcurrentSkipListMap<Long, String> concurrentSkipListMap = new ConcurrentSkipListMap<>();

        Instant start = Instant.now();


        // 消息确认成功回调
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            // 若是批量确认，删除已确认的消息，剩下的就是未确认的消息
            if (multiple) {
                ConcurrentNavigableMap<Long, String> confirmed = concurrentSkipListMap.headMap(deliveryTag);
                confirmed.clear();
            } else {
                concurrentSkipListMap.remove(deliveryTag);
            }
            System.out.println("确认消息tag：" + deliveryTag + "-" + multiple);
        };
        // 消息确认失败回调
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            String unConfirmMessage = concurrentSkipListMap.get(deliveryTag);
            System.out.println("未确认消息：" + unConfirmMessage + "未确认消息tag：" + deliveryTag);
        };
        // 消息监听器 监听消息的成功和失败
        channel.addConfirmListener(ackCallback, nackCallback);
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));
            // 记录所有要发送的消息
            concurrentSkipListMap.put(channel.getNextPublishSeqNo(), message);

        }

        Instant end = Instant.now();
        Duration duration = Duration.between(start, end);
        System.out.println("消息异步发布确认耗时：" + duration.toMillis() + "ms");
    }
}
